import getopt
import sys
import json

import pandas as pd
from datetime import datetime, timedelta
from kafka import KafkaProducer

from air_web.config.config import config
from air_web.dw.common_fun import create_topic
from air_web.data_platform import sql_engine


class StandardRate:

    def __init__(self, data_date):
        self.data_date = data_date if data_date else datetime.today().date()-timedelta(days=1)

    def get_diff(self, data_df):
        org_list = data_df['orgNo'].astype(str).tolist()

        sql = """select o.org_no as orgNo,
                         baseline_p, 
                         (baseline_p-p)/10000 as realAdjustAmount
                  from 
                  (select org_no, avg(p_total_sum) as p
                   from orgno_typeid_15min
                   where data_time>='{data_date}' and data_time < '{end_date}' 
                   and org_no in ({org_list}) and type_id=0
                   group by org_no) o
                  join(
                   select org_no,time_display_name, avg(p_total_sum) as baseline_p
                   from orgno_typeid_15min 
                   where data_time>='{baseline_date}' and data_time < '{data_date}' 
                   and org_no in ({org_list}) and type_id=0
                   group by org_no, time_display_name) bo
                  on o.org_no = bo.org_no
               """.format(org_list=','.join(org_list),
                          data_date=self.data_date,
                          end_date=self.data_date + timedelta(days=1),
                          baseline_date=self.data_date - timedelta(days=1),
                          )
        diff_df = sql_engine.query(sql)

        col_list = ['targetAdjustAmount', 'realAdjustAmount', 'adjustComplianceRate']
        if not diff_df.empty:
            data_df = pd.merge(data_df, diff_df, on='orgNo', how='left')
            data_df['targetAdjustAmount'] = data_df['baseline_p'] * data_df['control_prop']/10000
            data_df['adjustComplianceRate'] = (data_df['realAdjustAmount']/data_df[
                'targetAdjustAmount']*100)
            for col in col_list:
                data_df[col] = data_df[col].round(2)
        else:
            for col in col_list:
                data_df[col] = None
        for col in ['control_prop', 'baseline_p']:
            if col in data_df:
                data_df.drop(col, axis=1, inplace=True)
        return data_df

    def get_data(self):
        # 不推送执行时间范围后一天的
        sql = f"""select plan_id as planId, c.org_no as orgNo, control_time, r.org_level, 
                         control_prop
                  from city_approval_info c
                  left join real_org_no r on r.org_no=c.org_no
                  where control_time <= '{self.data_date}' 
                    and end_time >= '{self.data_date}'
               """
        approval_df = sql_engine.query(sql)
        if approval_df.empty:
            print(f"计划数据为空:{self.data_date}")
            return approval_df
        on5_df = approval_df.loc[approval_df['org_level'] == 1]
        on7_df = approval_df.loc[approval_df['org_level'] == 2]

        on5_list = on5_df['orgNo'].astype(str).tolist()
        on7_list = on7_df['orgNo'].astype(str).tolist()

        columns = ['planId', 'orgNo', 'complianceRate', 'executeTime', 'control_prop']
        data_df = pd.DataFrame()
        if on5_list:
            sql = f"""select on5 as orgNo,
                             control_time,
                             sum(n_cons_st)/sum(n_cons_total)*100 as complianceRate,
                             compute_date as executeTime
                      from city_control
                      where
                      compute_date='{self.data_date}'
                      and on5 in ({','.join(on5_list)})
                      group by on5, control_time
                   """
            df = sql_engine.query(sql)
            on5_df = on5_df.merge(df, on=['orgNo', 'control_time'], how='right')
            data_df = pd.concat([data_df, on5_df[columns]])
        if on7_list:
            sql = f"""select on7 as orgNo,
                                 control_time,
                                 n_cons_st/n_cons_total*100 as complianceRate,
                                 compute_date as executeTime
                          from city_control
                          where
                          compute_date='{self.data_date}'
                          and on7 in ({','.join(on7_list)})
                       """
            df = sql_engine.query(sql)
            on7_df = on7_df.merge(df, on=['orgNo', 'control_time'], how='right')
            data_df = pd.concat([data_df, on7_df[columns]])
        return data_df

    def main(self):
        data_df = self.get_data()
        if data_df.empty:
            print(f"达标率数据为空:{self.data_date}")
            return
        data_df = self.get_diff(data_df)

        data_df['complianceRate'] = data_df['complianceRate'].round(2)
        data_df['executeTime'] = pd.to_datetime(data_df['executeTime']).dt.strftime('%Y-%m-%d')
        data_df = data_df.astype(str)
        send_list = data_df.to_dict("record")

        producer = KafkaProducer(
            bootstrap_servers=','.join(config['KAFKA_HOST'])
        )
        topic_name = config['TOPIC_AIR_CONDITION_COMPLIANCE_RATE']
        create_topic(topic_name)
        for send_dict in send_list:
            message = {'data': send_dict, 'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
            value = json.dumps(message).encode('utf-8')
            producer.send(topic_name, value=value)
        # 刷新并关闭Kafka生产者
        producer.flush()
        producer.close()
        print(f"达标率推送kafka成功!num:{len(send_list)}")


if __name__ == "__main__":
    data_date = None

    opts, args = getopt.getopt(sys.argv[1:], "t:")
    for opt, val in opts:
        if opt == "-t":
            data_date = datetime.strptime(val, '%Y-%m-%d')

    StandardRate(data_date).main()
